package team.seekers.oj.websocket.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.socket.HandshakeInfo;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import team.seekers.oj.common.util.RedisUtils;
import team.seekers.oj.websocket.annotation.WebSocketMapping;
import team.seekers.oj.websocket.entity.MessageDTO;
import team.seekers.oj.websocket.entity.WebSocketSender;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;


@Component
@Slf4j
@WebSocketMapping("/ws/coding")
public class CodingHandler implements WebSocketHandler {

    public static final int MAX_LISTENING = 100;

    @Resource
    private ConcurrentHashMap<String, ConcurrentHashMap<String, WebSocketSender>> liveIdToSenderMap;

    @Resource
    private ConcurrentHashMap<String, WebSocketSender> senderMap;

    @Autowired
    private RedisUtils redisUtils;

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        HandshakeInfo handshakeInfo = session.getHandshakeInfo();
        InetSocketAddress remoteAddress = handshakeInfo.getRemoteAddress();
        log.info("[coding request] id: {}  from: {}", remoteAddress, session.getId());
        Mono<Void> output = session.send(Flux.create(sink -> {
            WebSocketSender sender = new WebSocketSender(session, sink);
            senderMap.put(sender.getId(), sender);
        })).then();

        Mono<Void> input = session.receive()
                .doOnNext(msg -> {
                    WebSocketSender sender = senderMap.get(session.getId());
//                    int size = sender.getListeningIdSet().size();
//                    // 限制最多监听 submission 数
//                    if (size >= MAX_LISTENING) {
//                        return;
//                    }
                    String messageText = msg.getPayloadAsText();

                    MessageDTO message = JSON.parseObject(messageText, new TypeReference<MessageDTO>() {});
                    if(message.getIsNew()){
                        String senderId = message.getMsg();
                        if(liveIdToSenderMap.get(senderId) == null){
                            sender.getListeningIdSet().add(senderId);
                            addNewWebSocketSender(senderId,sender);
                        }
                    }else {
                        log.info("[message] id: {}  msg: {}", remoteAddress, message.getMsg());
                        MessageDTO messageDTO = new MessageDTO();
                        messageDTO.setMsg("ok");
                        sender.sendData(messageDTO.toJson());
                    }
                })
                .doOnComplete(() -> {
                    // 浏览器端主动关闭
                    WebSocketSender sender = senderMap.get(session.getId());
                    for (String liveIdUserId : sender.getListeningIdSet()) {
                        if(liveIdToSenderMap.get(liveIdUserId) != null){
                            liveIdToSenderMap.remove(liveIdUserId);
                        }
                    }
                    senderMap.remove(session.getId());
                    log.info("doOnComplete");
                })
                .doOnCancel(() -> {

                    WebSocketSender sender = senderMap.get(session.getId());
                    for (String liveIdUserId : sender.getListeningIdSet()) {
                        if(liveIdToSenderMap.get(liveIdUserId) != null){
                            liveIdToSenderMap.remove(liveIdUserId);
                        }
                    }
                    senderMap.remove(session.getId());
                    log.info("doOnCancel");
                })
                .then();
        return Mono.zip(input, output).then();
    }


    private void addNewWebSocketSender(String renderId, WebSocketSender webSocketSender) {
        Map<String, WebSocketSender> webSocketSenders = liveIdToSenderMap
                .computeIfAbsent(renderId, k -> new ConcurrentHashMap<>());
        webSocketSenders.put(webSocketSender.getId(), webSocketSender);
        MessageDTO messageDTO = new MessageDTO();
        messageDTO.setMsg("ok");
        webSocketSender.sendData(messageDTO.toJson());
    }


    /**
     * 关闭该直播该用户写代码的权限
     * @param liveIdUserId
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "zzuoj.computer.class", durable = "true"),
            exchange = @Exchange(value = "zzuoj.computer.department", ignoreDeclarationExceptions = "true"),
            key = "chat"))
    public void unableCoding(String liveIdUserId) {
        System.out.println("关闭");
        if(liveIdToSenderMap.get(liveIdUserId) != null) {
            Collection<WebSocketSender> values = liveIdToSenderMap.get(liveIdUserId).values();
            for (WebSocketSender webSocketSender : values) {
                MessageDTO messageDTO = new MessageDTO();
                messageDTO.setMsg("close");
                webSocketSender.sendData(messageDTO.toJson());
            }
        }
    }
}